package host.skiree.nettydemo.client;

import host.skiree.nettydemo.client.handler.DemoRpcClientHandler;
import host.skiree.nettydemo.message.RpcRequestMessage;
import host.skiree.nettydemo.message.SequenceIdGenerator;
import host.skiree.nettydemo.protocol.MessageCodecSharable;
import host.skiree.nettydemo.service.HelloService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.Objects;

/**
 * @author gcl
 * @date 2022/9/27 9:13
 */
@Slf4j
public class NettyRpcClientManager {

    private static Channel channel = null;

    private static  final Object Lock = new Object();

    private static void initChannel() {
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            channel = new Bootstrap()
                    .group(workGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(10240, 11,4,1,0));
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                            ch.pipeline().addLast(new MessageCodecSharable());
                            ch.pipeline().addLast(new DemoRpcClientHandler());
                        }
                    })
                    .connect(new InetSocketAddress("127.0.0.1", 8088))
                    .sync()
                    .channel();
            channel.closeFuture().addListener(future -> {
                workGroup.shutdownGracefully();
            });
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 返回channel 双重检查锁
     * @return
     */
    public static Channel getChannel(){
        if (!Objects.isNull(channel)){
            return channel;
        }else {
            synchronized (Lock){
                if (!Objects.isNull(channel)){
                    return channel;
                }else {
                    initChannel();
                    return channel;
                }
            }
        }
    }

    public static void main(String[] args) {
        HelloService service = getProxyService(HelloService.class);
        System.out.println(service.sayHello("HelloZhangSan"));
        System.out.println(service.sayHello("HelloLiSi"));
        System.out.println(service.sayHello("HelloWangWu"));
        System.out.println(service.sayHello("HelloSunSan"));
    }

    /**
     * 使用JDK代理来发送消息，封装了请求对象的创建和发送的过程
     * https://blog.csdn.net/weixin_44042734/article/details/125819166
     * 这个只是发送数据，我们还要通过SequenceIdGenerator.nextId来在两个线程中同步数据
     */
    public static <T> T getProxyService(Class<T> serviceClass){
        ClassLoader loader = serviceClass.getClassLoader();
        Class<?>[] interfaces = new Class[]{serviceClass};
        Object o = Proxy.newProxyInstance(loader, interfaces, (proxy, method, args) -> {
            // 将方法的调用转换为RPC消息对象
            RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                    SequenceIdGenerator.nextId(),
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args
            );
            // 将消息发送出去
            getChannel().writeAndFlush(rpcRequestMessage);
            // 准备使用promise来接收结果                          指定promise异步接收结果的线程
            DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
            DemoRpcClientHandler.PROMISES_MAP.put(rpcRequestMessage.getSequenceId(), promise);
            // 等待promise的结果
            promise.await();
            // 返回结果
            if (promise.isSuccess()){
                // 调用正常
                return promise.getNow();
            }else {
                // 调用失败
                throw new RuntimeException(promise.cause());
            }
        });
        return (T)o;
    }

}
